package com.javaspeak.concurrency.structures.arrayblockingqueue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;

/**
 * @author John Dickerson
 */
public class ReadCallable implements Callable<Result> {

    private ArrayBlockingQueue<QueueObject> arrayBlockingQueue;

    public volatile boolean keepRunning = true;


    public ReadCallable( ArrayBlockingQueue<QueueObject> arrayBlockingQueue ){

        this.arrayBlockingQueue = arrayBlockingQueue;
    }


    @Override
    public Result call() {

        long nanoBeforeTime = System.nanoTime();

        int count = 0;

        @SuppressWarnings("unused")
        QueueObject queueObject = null;

        while ( keepRunning  ){

           try {
               queueObject = arrayBlockingQueue.take();
                count++;
           }
           catch( InterruptedException e ){

               keepRunning = false;
           }
        }

        while ( arrayBlockingQueue.poll() != null ){

            count++;
        }

        long processingNanoTime = System.nanoTime() - nanoBeforeTime;

        Result result =
            new Result( processingNanoTime, count,
                    Thread.currentThread().getName() );

        return result;
    }
}
